package org.example.using;

import static org.apache.spark.sql.functions.*;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.example.bean.TPhmGzzdYjgzbjRank;
import org.example.utils.MysqlUtils_2;

import java.io.IOException;


/***
 * 创建一个类,用来测试 spark 2.3.0版本的 DataFrame的用法, DataSet API的使用.
 */
public class Spark04_SparkDataSetJoinUsing {

    public static void main(String[] args) throws IOException {
        // 初始化SparkSession
        // 引入log4j,消除控制台的警告
        Logger.getLogger("org").setLevel(Level.WARN);
        Logger.getLogger("akka").setLevel(Level.WARN);
        SparkConf conf = new SparkConf().setAppName("PantographModelFunction") // 设置应用程序名称
                .setMaster("local[*]") // 设置运行模式，这里是本地模式
                .set("spark.executor.memory", "6g") // 设置executor内存大小
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 设置序列化器
                .set("spark.driver.allowMultipleContexts", "true") // 允许多个SparkContext同时存在
                .set("spark.hadoop.validateOutputSpecs", "false") // 禁止输出路径已经存在时抛出异常
                .set("hive.mapred.supports.subdirectories", "true") // 开启Hive支持
                .set("mapreduce.input.fileinputformat.input.dir.recursive", "true") // 开启递归读取文件夹中的文件
                .set("spark.debug.maxToStringFields", "100").set("spark.sql.codegen.wholeStage", "false")
                .set("spark.sql.caseSensitive", "false"); // Spark会自动处理大小写不匹配的情况
        JavaSparkContext sc = new JavaSparkContext(conf);
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        /*
            这个案例中,主要介绍了两个或者多个DataSet,进行 join,以及spark中的join方式
                在join中,关联键相同时怎么写,以及多个关联键怎么写,还有就是多张表关联后存在重复列的情况如何处理.
                给数据集中的列修改列名以及添加新的列等长碰到的需求实现.
         */
        Dataset<TPhmGzzdYjgzbjRank> tPhmGzzdYjgzbjRankDS = MysqlUtils_2.readMySQLTableDF(spark, "t_phm_gzzd_yjgzbj_rank", TPhmGzzdYjgzbjRank.class);
        tPhmGzzdYjgzbjRankDS.show();
        // 1、函数 lit()、withColumn() 函数的用法.在使用时,需要你手动导入: import static org.apache.spark.sql.functions.*; 这个包
        // lit() 函数在 Spark 中用于创建一个字面值（常量）列。当你需要在数据集中添加一个具有固定值的列时，可以使用 lit() 函数。以下是一个使用 lit() 函数的示例,这里搭配 `withColumn()`函数产生一个新列.
        tPhmGzzdYjgzbjRankDS.withColumn("new_column", lit("常量值")).show();

        // 2、介绍 withColumnRenamed() 函数,这个函数是为已经存在的列,修改列名称.不会产生新的字段,字段数量不变.
        tPhmGzzdYjgzbjRankDS.withColumnRenamed("sum_flag", "flag").show();
        System.out.println("==============");
        tPhmGzzdYjgzbjRankDS.withColumnRenamed("new_column", "new").show();

        // 3、`drop()`、`alias()` 函数的使用
        tPhmGzzdYjgzbjRankDS.drop("sum_flag").show(false); // 当前的数据集删除指定的列.从而达到减少列的目的,在做 join 时常用.
        // 这里是为当前的数据集取一个别名,通过别名删除指定的列,如果是对多个数据集进行操作,多个数据集中存在多个相同的列名称,这样很有必要.针对多个数据集进行join方式的时候会涉及到这种操作.
        tPhmGzzdYjgzbjRankDS.alias("t1").drop("t1.id").show(false); // 这样写,从显示结果上看,没啥效果.

        /*
            4、`Spark` 中的多表关联中涉及到的几种情况 `Inner Join (内连接)`、`Left Outer Join (左外连接)`、`Right Outer Join (右外连接)`、`Full Outer Join (全外连接)`、
                 `Left Semi Join (左半连接)`、`Left Anti Join (左反连接)`、`Cross Join (笛卡尔积连接)`
         */

        Dataset<Row> mxxqDS = MysqlUtils_2.readMysqlTableDs(spark, "t_phm_mxxq");
        // 4.1 join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "inner").show(false);
        // 4.2、full out join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "outer").show(false);
        // 4.3 left join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "left").show(false);
        // 4.4 right join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "right").show(false);
        // 4.5 left_semi join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "left_semi").show(false);
        // 4.6 left_anti join
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "left_anti").show(false);
        // 4.7 cross
        mxxqDS.alias("t1").join(tPhmGzzdYjgzbjRankDS.alias("t2"), expr("t1.xxms = t2.phm_gzmc"), "cross").show(false);
    }
}
